package com.arch.flume.source;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 生命周期方法: 构造器 -> configure -> start -> processor.process
 * 1.读取配置文件：（配置文件的内容：读取哪个文件、编码集、偏移量写到哪个文件、多长时间检查一下文件是否有新内容）
 * @author pizhihui
 * @date 2017-11-10
 */
public class TailFileSource extends AbstractSource implements EventDrivenSource, Configurable{

    private static final Logger logger = LoggerFactory.getLogger(TailFileSource.class);

    private String filePath;
    private String charset;
    private String positionFile;
    private Long interval;
    private ExecutorService executor;
    private FileRunnable fileRunnable;

    @Override
    public void configure(Context context) {
        filePath = context.getString("filePath");
        charset = context.getString("charset", "UTF-8");
        positionFile = context.getString("positionFile");
        interval = context.getLong("interval", 1000L);
    }

    @Override
    public synchronized void start() {
        executor = Executors.newSingleThreadExecutor();
        fileRunnable = new FileRunnable(filePath, charset, positionFile, interval, getChannelProcessor());
        executor.submit(fileRunnable);
        super.start();
    }

    @Override
    public synchronized void stop() {
        fileRunnable.setFlag(false);
        executor.shutdown();
        while (!executor.isTerminated()) {
            logger.debug("Waiting for filer executor service to stop");
            try {
                executor.awaitTermination(500, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                logger.debug("Interrupted while waiting for exec executor service "
                        + "to stop. Just exiting.");
                Thread.currentThread().interrupt();
            }
        }
        super.stop();
    }


    // 私有静态内部类:
    private static class FileRunnable implements Runnable {

        private String filePath;
        private String charset;
        private String positionFile;
        private Long interval;
        private Long offset = 0L;
        private final ChannelProcessor channelProcessor;
        private RandomAccessFile raf;
        private File posFile;
        private boolean flag = true;

        public FileRunnable(String filePath, String charset, String positionFile, Long interval, ChannelProcessor channelProcessor) {
            this.filePath = filePath;
            this.charset = charset;
            this.positionFile = positionFile;
            this.interval = interval;
            this.channelProcessor = channelProcessor;

            // 先读偏移量,如果有,安装偏移量的值接着读,否则从头开始读
            posFile = new File(positionFile);
            if (!posFile.exists()) {
                try {
                    posFile.createNewFile();
                } catch (IOException e) {
                    logger.error("create position file error: {}", e);
                }
            }


            try {
                // 读取偏移量
                String fileOffsetString = FileUtils.readFileToString(posFile);
                if (fileOffsetString != null && !"".equals(fileOffsetString)) {
                    // 获取到偏移量
                    offset = Long.parseLong(fileOffsetString);
                }

                // 从指定的偏移量读取日志文件
                raf = new RandomAccessFile(filePath, "r");
                raf.seek(offset);

            } catch (IOException e) {
                logger.error("read position file error: {}", e);
            }

        }

        @Override
        public void run() {
            while (flag) {
                try {
                    String line = raf.readLine();
                    if (line != null && "".equals(line)) {
                        line = new String(line.getBytes("ISO-8859-1"), charset);
                        channelProcessor.processEvent(EventBuilder.withBody(line.getBytes()));
                        offset = raf.getFilePointer();
                        FileUtils.writeStringToFile(posFile, offset + "");
                    } else {
                        TimeUnit.SECONDS.sleep(interval);
                    }
                } catch (IOException e) {
                    logger.error("read log file error: ", e);
                } catch (InterruptedException e) {
                    logger.error("thread sleep error: ", e);
                }
            }

        }

        private void setFlag(boolean flag) {
            this.flag = flag;
        }
    }


}
